package qc.module.qms.service.rawdata;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import qc.common.core.enums.CommunicationChannelTypeEnum;
import qc.common.core.pair.KeyValuePairDto;
import qc.common.core.utils.LocalDateTimeUtil;
import qc.common.core.utils.SqlRunnerUtil;
import qc.common.core.utils.StringSplitUtil;
import qc.module.platform.dto.db.DBConnectionDto;
import qc.module.qms.dto.rawdata.QmsRawDataOperateResultDto;
import qc.module.qms.dto.rawdata.QmsRawDataReceiveDto;
import qc.module.qms.dto.rawdata.QmsRawDataSyncResultDto;
import qc.module.qms.entity.QmsRawdataSync;
import qc.module.qms.repository.QmsRawdataSyncRepository;

import java.sql.SQLException;
import java.time.LocalDateTime;
import java.util.*;

/**
 * QmsRawdataSyncService
 *
 * @author QuCheng Tech
 * @since 2025/11/23
 */
@Service
public class QmsRawdataSyncService {
    private static final Logger logger = LoggerFactory.getLogger(QmsRawdataSyncService.class);

    @Autowired
    private QmsRawdataSyncRepository repository;

    @Autowired
    private QmsRawDataRecordService rawDataRecordService;

    /**
     * 定时任务自动同步原始数据，根据原始数据同步记录中的上次同步记录继续进行同步
     *
     * @param rawdataSync 同步entity
     * @param srcCon      来源数据库连接
     * @param destCon     目标数据库连接
     * @return qc.module.qms.dto.rawdata.QmsRawDataSyncResultDto
     * @author QuCheng Tech
     * @since 2025/11/24
     */
    public QmsRawDataSyncResultDto autoSyncDatas(QmsRawdataSync rawdataSync, DBConnectionDto srcCon, DBConnectionDto destCon) {
        return syncDatas(rawdataSync, srcCon, destCon, false, null, null);
    }

    /**
     * 人工操作进行原始数据同步，可以指定开始和截止控制同步字段的值用于控制同步的数据范围
     *
     * @param rawdataSync       同步entity
     * @param srcCon            来源数据库连接
     * @param destCon           目标数据库连接
     * @param beginControlValue 起始控制值，不能为空
     * @param endControlValue   截止控制值，可以为空
     * @return qc.module.qms.dto.rawdata.QmsRawDataSyncResultDto
     * @author QuCheng Tech
     * @since 2025/11/24
     */
    public QmsRawDataSyncResultDto manualSyncDatas(QmsRawdataSync rawdataSync, DBConnectionDto srcCon, DBConnectionDto destCon, String beginControlValue, String endControlValue) {
        return syncDatas(rawdataSync, srcCon, destCon, true, beginControlValue, endControlValue);
    }

    /**
     * 原始数据同步处理
     *
     * @param rawdataSync       同步entity
     * @param srcCon            来源数据库连接
     * @param destCon           目标数据库连接
     * @param isManual          是否为人工操作进行同步
     * @param beginControlValue 起始控制值，不能为空
     * @param endControlValue   截止控制值，可以为空
     * @return qc.module.qms.dto.rawdata.QmsRawDataSyncResultDto
     * @author QuCheng Tech
     * @since 2025/11/24
     */
    QmsRawDataSyncResultDto syncDatas(QmsRawdataSync rawdataSync, DBConnectionDto srcCon, DBConnectionDto destCon, boolean isManual, String beginControlValue, String endControlValue) {
        //先初始化返回结果，设置是否成功标记为true
        QmsRawDataSyncResultDto result = new QmsRawDataSyncResultDto();
        result.setSuccess(true);

        //输入信息在方法调用前已经通过验证，此处不再重复进行验证
        String srcTableCode = rawdataSync.getTablecode();
        String destTableCode = rawdataSync.getDesttablecode();
        boolean isControlByTime = false;
        if (rawdataSync.getCtrlbytm() != null && rawdataSync.getCtrlbytm().booleanValue() == true)
            isControlByTime = true;
        result.setControlByTime(isControlByTime);
        //使用SqlRunnerUtil查询出的结果中所有列名称均已经处理为小写字母，在此获取对应列名时也全部处理为小写字母
        String controlColumn = rawdataSync.getCtrlcolumn().toLowerCase();
        String dataTimeColumn = rawdataSync.getTmcolumn().toLowerCase();
        String channelTypeColumn = null;
        if (StringUtils.isNotBlank(rawdataSync.getChlcolumn()))
            channelTypeColumn = rawdataSync.getChlcolumn().toLowerCase();
        Map<String, CommunicationChannelTypeEnum> channelMaps = null;
        //站点编码列名，多个使用逗号分隔
        String[] stcdColumns = StringSplitUtil.Split(rawdataSync.getStcdcolumns().toLowerCase(), true);
        //数据值列名，多个使用逗号分隔
        String[] dataValueColumns = StringSplitUtil.Split(rawdataSync.getDvcolumns().toLowerCase(), true);
        //目标数据值列名，多个使用逗号分隔，不设置时保持与来源列名相同
        String[] destDataValueColumns = null;
        if (StringUtils.isNotBlank(rawdataSync.getDestdvcolumns()))
            destDataValueColumns = StringSplitUtil.Split(rawdataSync.getDestdvcolumns().toLowerCase(), true);
        //如果目标数值列为空或者长度小于来源数值列，默认使用来源数值列集合
        if (destDataValueColumns == null || destDataValueColumns.length < dataValueColumns.length)
            destDataValueColumns = dataValueColumns;

        //获取来源通道类型与目标原始数据表中的通道类型对应关系
        if (StringUtils.isNotBlank(channelTypeColumn) && StringUtils.isNotBlank(rawdataSync.getChlmap())) {
            String[] strs = StringSplitUtil.Split(rawdataSync.getChlmap(), false);
            if (strs != null && strs.length > 0x0) {
                channelMaps = new LinkedHashMap<>();
                //对应关系按照CommunicationChannelTypeEnum枚举的顺序
                CommunicationChannelTypeEnum[] enums = CommunicationChannelTypeEnum.values();
                for (int i = 0; i < enums.length; i++) {
                    String srcChannelString = strs[i];
                    if (StringUtils.isNotBlank(srcChannelString) && channelMaps.containsKey(srcChannelString) == false)
                        channelMaps.put(srcChannelString, enums[i]);
                }
            }
        }

        //源数据库连接信息
        String dbUrl = srcCon.getUrl();
        String dbUser = srcCon.getUser();
        String dbPassword = srcCon.getPassword();

        //根据不同的操作使用不同的查询sql进行查询待同步的原始数据
        List<Map<String, Object>> queryData = null;
        if (isManual == false) {
            //如果是自动定时任务进行同步操作，获取上次的控制字段的值，如果没有则不需要条件
            if (StringUtils.isBlank(rawdataSync.getSyncctrlvalue())) {
                String sql = "SELECT * FROM " + srcTableCode + " ORDER BY " + controlColumn + " ASC";
                try {
                    queryData = SqlRunnerUtil.executeSelectListSql(dbUrl, dbUser, dbPassword, sql);
                } catch (SQLException ex) {
                    String errorMsg = "查询待同步的原始数据时sql语句异常，" + ex.getMessage();
                    logger.error(errorMsg);
                    result.setSuccess(false);
                    result.setMsg(errorMsg);
                    return result;
                }
            } else {
                Object latestControlValue = rawdataSync.getSyncctrlvalue();
                //将上次的控制字段的值转换为对应的类型
                if (isControlByTime == true) {
                    //根据时间进行同步控制
                    latestControlValue = LocalDateTimeUtil.parse(rawdataSync.getSyncctrlvalue());
                    if (latestControlValue == null) {
                        //如果转换上次获取同步数据的时间为null，无法判断本次要同步哪些数据，进行错误提示
                        String errorMsg = "系统定时进行原始数据同步处理时根据上次同步数据时间[" + rawdataSync.getSyncctrlvalue() + "]无法得到正确的时间，不能继续进行数据同步处理。";
                        logger.error(errorMsg);
                        result.setSuccess(false);
                        result.setMsg(errorMsg);
                        return result;
                    }
                } else {
                    //根据自增数值控制同步
                    int intValue = Integer.parseInt(rawdataSync.getSyncctrlvalue());
                    if (intValue <= 0x0) {
                        //如果转换上次获取同步数据的数值无效，无法判断本次要同步哪些数据，进行错误提示
                        String errorMsg = "系统定时进行原始数据同步处理时根据上次同步数据数值[" + rawdataSync.getSyncctrlvalue() + "]无法得到正确的数值，不能继续进行数据同步处理。";
                        logger.error(errorMsg);
                        result.setSuccess(false);
                        result.setMsg(errorMsg);
                        return result;
                    } else {
                        latestControlValue = intValue;
                    }
                }

                String sql = "SELECT * FROM " + srcTableCode + " WHERE " + controlColumn + " >?" + " ORDER BY " + controlColumn + " ASC";
                try {
                    queryData = SqlRunnerUtil.executeSelectListSql(dbUrl, dbUser, dbPassword, sql, latestControlValue);
                } catch (SQLException ex) {
                    String errorMsg = "查询待同步的原始数据时sql语句异常，" + ex.getMessage();
                    logger.error(errorMsg);
                    result.setSuccess(false);
                    result.setMsg(errorMsg);
                    return result;
                }
            }
        } else {
            //如果是人工触发的同步操作，使用传入的起始控制值
            Object beginValue = null;
            Object endValue = null;
            if (isControlByTime == true) {
                //根据时间进行同步控制
                LocalDateTime time = LocalDateTimeUtil.parse(beginControlValue);
                if (time != null)
                    beginValue = time;
                time = LocalDateTimeUtil.parse(endControlValue);
                if (time != null)
                    endValue = time;
            } else {
                //根据自增数值控制同步
                int intValue = Integer.parseInt(beginControlValue);
                if (intValue > 0x0)
                    beginValue = intValue;
                intValue = Integer.parseInt(endControlValue);
                if (intValue > 0x0)
                    endValue = intValue;
            }
            //开始的值不能为空，为空时无法进行同步
            if (beginValue == null) {
                //进行错误提示
                String errorMsg = "人工操作进行原始数据同步处理时根据同步起始值[" + beginControlValue + "]无法转换为有效的值，不能进行数据同步处理。";
                logger.error(errorMsg);
                result.setSuccess(false);
                result.setMsg(errorMsg);
                return result;
            } else {
                //需要区分是否有截止的值，sql语句和参数不同
                if (endValue == null) {
                    try {
                        String sql = "SELECT * FROM " + srcTableCode + " WHERE " + controlColumn + " >=?" + " ORDER BY " + controlColumn + " ASC";
                        queryData = SqlRunnerUtil.executeSelectListSql(dbUrl, dbUser, dbPassword, sql, beginValue);
                    } catch (SQLException ex) {
                        String errorMsg = "查询待同步的原始数据时sql语句异常，" + ex.getMessage();
                        logger.error(errorMsg);
                        result.setSuccess(false);
                        result.setMsg(errorMsg);
                        return result;
                    }
                } else {
                    String sql = "SELECT * FROM " + srcTableCode + " WHERE " + controlColumn + " >=?" + " AND " + controlColumn + " <=?" + " ORDER BY " + controlColumn + " ASC";
                    try {
                        queryData = SqlRunnerUtil.executeSelectListSql(dbUrl, dbUser, dbPassword, sql, beginValue, endValue);
                    } catch (SQLException ex) {
                        String errorMsg = "查询待同步的原始数据时sql语句异常，" + ex.getMessage();
                        logger.error(errorMsg);
                        result.setSuccess(false);
                        result.setMsg(errorMsg);
                        return result;
                    }
                }
            }
        }
        //判断查询出的源数据记录
        if (queryData == null || queryData.size() < 0x1) {
            //没有需要同步的数据
            result.setSuccess(true);
            result.setMsg("来源数据表中没有需要进行同步的数据。");
            result.setCount(0x0);
            result.setBeginControlValue(null);
            result.setEndControlValue(null);
            result.setBeginDataTime(null);
            result.setEndDataTime(null);
        } else {
            //有需要同步的数据，逐条遍历进行处理
            List<QmsRawDataReceiveDto> rawdatas = new LinkedList<>();

            //记录本次同步数据中的起止数据ID/时间
            int count = 0x0;
            LocalDateTime syncBeginDataTime = null;
            LocalDateTime syncEndDataTime = null;
            LocalDateTime syncBeginControlTime = null;
            LocalDateTime syncEndControlTime = null;
            int syncBeginControlIntValue = 0x0;
            int syncEndControlIntValue = 0x0;

            for (Map<String, Object> dataRow : queryData) {
                count++;//同步的来源数据数量加1

                QmsRawDataReceiveDto rawData = new QmsRawDataReceiveDto();
                rawData.setTablecode(destTableCode);

                //获取站点编码、时标、通道类型
                //站点编码可能是多列，如果是多列使用列1-列2-列n依次排列合并为一个
                String stcd = dataRow.get(stcdColumns[0x0]).toString();
                if (stcdColumns.length > 0x1) {
                    for (int i = 1; i < stcdColumns.length; i++) {
                        stcd += "-" + dataRow.get(stcdColumns[i]).toString();
                    }
                }
                rawData.setStcd(stcd);
                //获取数据时标
                String rowDataTimeString = dataRow.get(dataTimeColumn).toString();
                rawData.setTm(rowDataTimeString);
                LocalDateTime rowDataTime = LocalDateTimeUtil.parse(rowDataTimeString);
                if (rowDataTime != null) {
                    if (syncBeginDataTime == null || syncBeginDataTime.isAfter(rowDataTime))
                        syncBeginDataTime = rowDataTime;
                    if (syncEndDataTime == null || syncEndDataTime.isBefore(rowDataTime))
                        syncEndDataTime = rowDataTime;
                }
                //如果没有通道类型列，使用默认值
                rawData.setChannel(CommunicationChannelTypeEnum.UN_DEFINE.getIndex());
                if (StringUtils.isNotBlank(channelTypeColumn) && channelMaps != null && channelMaps.size() > 0x0) {
                    String srcChannelString = dataRow.get(channelTypeColumn).toString();
                    if (channelMaps.containsKey(srcChannelString))
                        rawData.setChannel(channelMaps.get(srcChannelString).getIndex());
                }
                //获取用于同步控制字段的数值/时间，判断出本次同步的起始和截止值
                String controlFieldValueString = dataRow.get(controlColumn).toString();
                if (StringUtils.isNotBlank(controlFieldValueString)) {
                    if (isControlByTime == true) {
                        LocalDateTime controlFieldTime = LocalDateTimeUtil.parse(controlFieldValueString);
                        if (controlFieldTime != null) {
                            if (syncBeginControlTime == null || syncBeginControlTime.isAfter(controlFieldTime))
                                syncBeginControlTime = controlFieldTime;
                            if (syncEndControlTime == null || syncEndControlTime.isBefore(controlFieldTime))
                                syncEndControlTime = controlFieldTime;
                        }
                    } else {
                        try {
                            int controlFiledInt = Integer.parseInt(controlFieldValueString);
                            if (controlFiledInt > 0x0) {
                                if (syncBeginControlIntValue == 0x0 || syncBeginControlIntValue > controlFiledInt)
                                    syncBeginControlIntValue = controlFiledInt;
                                if (syncEndControlIntValue == 0x0 || syncEndControlIntValue < controlFiledInt)
                                    syncEndControlIntValue = controlFiledInt;
                            }
                        } catch (NumberFormatException ex) {

                        }
                    }
                }

                //获取数据值列集合
                List<KeyValuePairDto> elements = new LinkedList<>();
                for (int i = 0; i < dataValueColumns.length; i++) {
                    String fieldValue = dataRow.get(dataValueColumns[i]).toString();

                    KeyValuePairDto kv = new KeyValuePairDto();
                    kv.setKey(destDataValueColumns[i]);
                    kv.setValue(fieldValue);

                    elements.add(kv);
                }

                rawData.setElements(elements);

                rawdatas.add(rawData);
            }

            //调用原始数据记录Service中的接收方法，由接收方法进行发送至MQ或保存至数据库处理
            HashMap<String, DBConnectionDto> connections = new HashMap<>();
            connections.put(destTableCode, destCon);
            QmsRawDataOperateResultDto receiveResult = rawDataRecordService.receive(rawdatas, connections);
            if (receiveResult.isSuccess() == true) {
                //设置返回结果
                result.setSuccess(true);
                result.setCount(count);
                if (isControlByTime == true) {
                    result.setBeginControlValue(syncBeginControlTime);
                    result.setEndControlValue(syncEndControlTime);
                } else {
                    result.setBeginControlValue(syncBeginControlIntValue);
                    result.setEndControlValue(syncEndControlIntValue);
                }
                result.setBeginDataTime(syncBeginDataTime);
                result.setEndDataTime(syncEndDataTime);
            } else {
                //调用原始数据接收接口返回失败
                result.setSuccess(false);
                result.setMsg(receiveResult.getMsg());
                result.setCount(0x0);
                result.setBeginControlValue(null);
                result.setEndControlValue(null);
                result.setBeginDataTime(null);
                result.setEndDataTime(null);
            }
        }

        return result;
    }

    /**
     * 更新指定的数据同步的最新同步处理时间
     *
     * @param tableCode 数据库表编码
     * @param time      数据同步的处理时间
     * @param msg       数据同步的处理结果信息
     * @return void
     * @author QuCheng Tech
     * @since 2025/11/24
     */
    public void updateLatestSyncHandleTime(String tableCode, LocalDateTime time, String msg) {
        try {
            if (StringUtils.isNotBlank(tableCode)) {
                //如果传入的处理时间为空，默认使用当前时间
                if (time == null)
                    time = LocalDateTimeUtil.getNow();

                LambdaUpdateWrapper<QmsRawdataSync> wrapper = new LambdaUpdateWrapper<>();
                wrapper.set(QmsRawdataSync::getSynctm, time);
                wrapper.set(QmsRawdataSync::getSyncmsg, msg);

                wrapper.eq(QmsRawdataSync::getTablecode, tableCode);

                repository.update(null, wrapper);
            }
        } catch (Exception ex) {
            logger.error("更新原始数据同步的最新同步处理时间异常，" + ex.getMessage());
        }
    }

    /**
     * 更新指定的数据同步的最新完成同步的数据时标、同步控制使用的记录ID/时间
     *
     * @param tableCode    数据库表编码
     * @param time         数据同步的处理时间
     * @param msg          数据同步的处理结果信息
     * @param dataTime     数据时标
     * @param controlValue 控制同步值
     * @return void
     * @author QuCheng Tech
     * @since 2025/11/24
     */
    public void updateLatestSyncDataTime(String tableCode, LocalDateTime time, String msg, LocalDateTime dataTime, String controlValue) {
        try {
            if (StringUtils.isNotBlank(tableCode)) {
                //如果传入的处理时间为空，默认使用当前时间
                if (time == null)
                    time = LocalDateTimeUtil.getNow();

                LambdaUpdateWrapper<QmsRawdataSync> wrapper = new LambdaUpdateWrapper<>();
                wrapper.set(QmsRawdataSync::getSynctm, time);
                wrapper.set(QmsRawdataSync::getSyncmsg, msg);
                wrapper.set(QmsRawdataSync::getSyncdatatm, dataTime);
                wrapper.set(QmsRawdataSync::getSyncctrlvalue, controlValue);

                wrapper.eq(QmsRawdataSync::getTablecode, tableCode);

                repository.update(null, wrapper);
            }
        } catch (Exception ex) {
            logger.error("更新原始数据同步的最新同步数据时标异常，" + ex.getMessage());
        }
    }

    /**
     * 验证指定的原始数据同步entity配置是否正确
     *
     * @param syncEntity 原始数据同步entity
     * @return qc.module.qms.dto.rawdata.QmsRawDataSyncResultDto
     * @author QuCheng Tech
     * @since 2025/11/24
     */
    public QmsRawDataSyncResultDto validRawDataSyncConfig(QmsRawdataSync syncEntity) {
        //先初始化返回结果，设置是否成功标记为true
        QmsRawDataSyncResultDto result = new QmsRawDataSyncResultDto();
        result.setSuccess(true);

        //根据数据库表编码获取同步配置entity，不能为空
        if (syncEntity == null) {
            result.setMsg("进行数据同步的来源数据库表信息不存在");

            return result;
        }
        if (StringUtils.isBlank(syncEntity.getTablecode())) {
            result.setSuccess(false);
            result.setMsg("进行数据同步的来源数据库表编码不能为空");

            return result;
        }
        //基本配置信息验证：同步控制列名、站点编码列名、数据时标列名、数据值列名、目标数据库表编码必须设置并且有效
        if (StringUtils.isBlank(syncEntity.getCtrlcolumn())) {
            result.setSuccess(false);
            result.setMsg("进行数据同步的同步控制列名不能为空");

            return result;
        }
        if (StringUtils.isBlank(syncEntity.getStcdcolumns())) {
            result.setSuccess(false);
            result.setMsg("进行数据同步的站点编码列名不能为空");

            return result;
        }
        if (StringUtils.isBlank(syncEntity.getTmcolumn())) {
            result.setSuccess(false);
            result.setMsg("进行数据同步的数据时标列名不能为空");

            return result;
        }
        if (StringUtils.isBlank(syncEntity.getDvcolumns())) {
            result.setSuccess(false);
            result.setMsg("进行数据同步的数据值列名不能为空");

            return result;
        }
        if (StringUtils.isBlank(syncEntity.getDesttablecode())) {
            result.setSuccess(false);
            result.setMsg("进行数据同步的目标数据库表编码不能为空");

            return result;
        }

        return result;
    }

    /**
     * 获取指定的数据库表编码的原始数据同步Entity
     *
     * @param tableCode 数据库表编码
     * @return qc.module.qms.entity.QmsRawdataSync
     * @author QuCheng Tech
     * @since 2025/11/24
     */
    public QmsRawdataSync getEntity(String tableCode) {
        if (StringUtils.isNotBlank(tableCode)) {
            LambdaQueryWrapper<QmsRawdataSync> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(QmsRawdataSync::getTablecode, tableCode);

            return repository.selectOne(wrapper);
        }

        return null;
    }
}
